管理设计篇之"分布式锁" [2026重制版]

# 加锁：仅在 key 不存在时设置，并设置过期时间
SET resource_name my_random_value NX PX 30000

-- 比对 value，确保只有锁的持有者才能释放
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.27.0</version>
</dependency>

@Configuration
public class RedissonConfig {

    @Bean(destroyMethod = "shutdown")
    public RedissonClient redissonClient() {
        Config config = new Config();
        // 单节点模式
        config.useSingleServer()
              .setAddress("redis://127.0.0.1:6379")
              .setPassword("your_password")
              .setDatabase(0)
              .setConnectionPoolSize(64)
              .setConnectionMinimumIdleSize(24)
              .setIdleConnectionTimeout(10000)
              .setConnectTimeout(5000)
              .setTimeout(3000)
              .setRetryAttempts(3)
              .setRetryInterval(1500);
        return Redisson.create(config);
    }

    // 集群模式（推荐生产使用）
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redissonClusterClient() {
        Config config = new Config();
        config.useClusterServers()
              // 集群节点列表
              .addNodeAddress("redis://node1:6379",
                             "redis://node2:6379",
                             "redis://node3:6379")
              .setPassword("your_password")
              .setScanInterval(5000); // 集群状态扫描间隔
        return Redisson.create(config);
    }
}

@Service
public class InventoryService {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 扣减库存 - 使用可重入锁
     * @param 商品ID
     * @param 扣减数量
     * @return 是否成功
     */
    public boolean deductStock(String productId, int quantity) {
        // 定义锁的名称（建议包含业务标识）
        String lockKey = "lock:inventory:" + productId;
        RLock lock = redissonClient.getLock(lockKey);

        try {
            // 尝试加锁，最多等待 10 秒，锁定后 30 秒自动释放
            // 注意：指定了 leaseTime 时 watchdog 机制不会启用（见下文说明）
            boolean acquired = lock.tryLock(10, 30, TimeUnit.SECONDS);

            if (!acquired) {
                log.warn("获取库存锁失败，商品ID: {}", productId);
                return false;
            }

            // ====== 业务逻辑开始 ======
            // 1. 查询库存
            Integer currentStock = getStockFromDB(productId);

            if (currentStock < quantity) {
                log.info("库存不足，商品ID: {}，当前库存: {}", productId, currentStock);
                return false;
            }

            // 2. 扣减库存
            boolean success = updateStockInDB(productId, currentStock - quantity);

            if (success) {
                log.info("扣减库存成功，商品ID: {}，数量: {}", productId, quantity);
            }
            // ====== 业务逻辑结束 ======

            return success;

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("获取锁被中断", e);
            return false;
        } finally {
            // 仅当锁是当前线程持有的才释放
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.5.0</version>
</dependency>

@Service
public class ZkDistributedLockService {

    private final InterProcessMutex lock;

    public ZkDistributedLockService(CuratorFramework client) {
        // 创建互斥锁，路径为 /locks/my-lock
        this.lock = new InterProcessMutex(client, "/locks/my-lock");
    }

    public void doWithLock() {
        try {
            // 尝试获取锁，最多等待 10 秒
            boolean acquired = lock.acquire(10, TimeUnit.SECONDS);

            if (acquired) {
                try {
                    // 执行业务逻辑
                    executeBusinessLogic();
                } finally {
                    lock.release();
                }
            } else {
                log.warn("获取 ZooKeeper 锁超时");
            }
        } catch (Exception e) {
            log.error("ZooKeeper 锁异常", e);
        }
    }
}

// 使用 JetCD 客户端（推荐）
// Maven: io.etcd:jetcd-core:0.7.6

@Service
public class EtcdDistributedLockService {

    private final Client etcdClient;
    private final Lease leaseClient;
    private final Lock lockClient;

    public EtcdDistributedLockService() {
        this.etcdClient = Client.builder()
                .endpoints("http://etcd1:2379", "http://etcd2:2379")
                .build();
        this.leaseClient = etcdClient.getLeaseClient();
        this.lockClient = etcdClient.getLockClient();
    }

    public void doWithLock(String lockName) {
        try {
            // 获取一个 30 秒的租约
            LeaseGrantResponse leaseResp = leaseClient.grant(30).get();
            long leaseId = leaseResp.getID();

            // 使用租约创建锁
            LockResponse lockResp = lockClient.lock(
                    ByteSequence.from(lockName, StandardCharsets.UTF_8),
                    leaseId
            ).get();

            try {
                // 业务逻辑
                businessLogic();
            } finally {
                // 解锁
                lockClient.unlock(lockResp.getKey()).get();
            }
        } catch (Exception e) {
            log.error("etcd 锁操作异常", e);
        }
    }
}

/**
 * 库存扣减服务 - 生产级实现
 *
 * 设计要点：
 * 1. 使用 Redisson 可重入锁
 * 2. 双层校验（缓存 + 数据库 CAS）
 * 3. 异步扣减，提升吞吐
 * 4. 完善的日志与监控
 */
@Service
@Slf4j
public class StockDeductionServiceImpl implements StockDeductionService {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    /** 锁等待时间（秒） */
    private static final long LOCK_WAIT_TIME = 5;
    /** 锁持有时间（秒） */
    private static final long LOCK_LEASE_TIME = 10;
    /** 库存缓存 Key 前缀 */
    private static final String STOCK_CACHE_PREFIX = "stock:";

    @Override
    public StockDeductResult deduct(String productId, int quantity, String orderId) {
        long startTime = System.currentTimeMillis();

        // 1. 快速路径：检查缓存中的库存是否充足
        String cacheKey = STOCK_CACHE_PREFIX + productId;
        String stockStr = stringRedisTemplate.opsForValue().get(cacheKey);

        if (stockStr != null) {
            int cachedStock = Integer.parseInt(stockStr);
            if (cachedStock < quantity) {
                return StockDeductResult.fail("库存不足", cachedStock);
            }
        }

        // 2. 获取分布式锁
        String lockKey = "lock:stock:" + productId;
        RLock lock = redissonClient.getLock(lockKey);

        try {
            boolean locked = lock.tryLock(
                    LOCK_WAIT_TIME,
                    LOCK_LEASE_TIME,
                    TimeUnit.SECONDS
            );

            if (!locked) {
                log.warn("获取库存锁失败，productId: {}, orderId: {}", productId, orderId);
                return StockDeductResult.fail("系统繁忙，请稍后重试", -1);
            }

            // 3. 二次校验库存（双重检查）
            stockStr = stringRedisTemplate.opsForValue().get(cacheKey);
            int currentStock = (stockStr != null) ? Integer.parseInt(stockStr)
                                                  : getStockFromDB(productId);

            if (currentStock < quantity) {
                // 刷新缓存
                refreshStockCache(productId, currentStock);
                return StockDeductResult.fail("库存不足", currentStock);
            }

            // 4. 数据库 CAS 更新（乐观锁）
            int affected = productMapper.decreaseStockWithCAS(productId, quantity, currentStock);

            if (affected > 0) {
                // 5. 更新缓存
                int newStock = currentStock - quantity;
                stringRedisTemplate.opsForValue().set(cacheKey, String.valueOf(newStock), 1, TimeUnit.HOURS);

                // 6. 发布事件（异步处理后续逻辑）
                eventPublisher.publishEvent(new StockDeductedEvent(productId, quantity, orderId));

                long cost = System.currentTimeMillis() - startTime;
                log.info("库存扣减成功，productId: {}, orderId: {}, 耗时: {}ms", productId, orderId, cost);

                return StockDeductResult.success(newStock);
            } else {
                // CAS 失败，说明有并发修改
                log.warn("库存扣减CAS失败，productId: {}", productId);
                return StockDeductResult.fail("库存变化，请重试", -1);
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return StockDeductResult.fail("操作中断", -1);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    /**
     * 从数据库查询库存（带兜底）
     */
    private int getStockFromDB(String productId) {
        Product product = productMapper.selectById(productId);
        if (product == null) {
            throw new BusinessException("商品不存在");
        }
        return product.getStock();
    }

    /**
     * 刷新库存缓存
     */
    private void refreshStockCache(String productId, int stock) {
        stringRedisTemplate.opsForValue().set(
                STOCK_CACHE_PREFIX + productId,
                String.valueOf(stock),
                1, TimeUnit.HOURS
        );
    }
}

@Mapper
public interface ProductMapper {

    /**
     * 使用乐观锁扣减库存
     * UPDATE ... SET stock = stock - #{quantity} WHERE id = #{id} AND stock = #{expectedStock}
     */
    @Update("UPDATE product SET stock = stock - #{quantity}, " +
            "update_time = NOW() " +
            "WHERE id = #{productId} AND stock >= #{quantity} AND stock = #{expectedStock}")
    int decreaseStockWithCAS(@Param("productId") String productId,
                              @Param("quantity") int quantity,
                              @Param("expectedStock") int expectedStock);
}